package com.tico.editor.editors.kafkaThread;

import com.tico.editor.editors.domain.Kafka;
import com.tico.editor.editors.domain.KafkaTopic;
import com.tico.editor.editors.service.HttpEditorService;
import com.tico.editor.editors.service.KafkaService;
import com.tico.editor.editors.service.KafkaTopicService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;

public class KafkaThread implements Runnable{
    public static List<String> kafkaTopicList;
    public static Properties props = null;
    public static  KafkaConsumer<String, String> consumer;
    final static ConcurrentLinkedQueue<String> subscribedTopics = new ConcurrentLinkedQueue<>();
    private KafkaService kafkaService;
    private KafkaTopicService kafkaTopicService;
    private HttpEditorService httpEditorService;

    public KafkaThread(KafkaService kafkaService, KafkaTopicService kafkaTopicService, HttpEditorService httpEditorService) {
        this.kafkaService = kafkaService;
        this.kafkaTopicService = kafkaTopicService;
        this.httpEditorService = httpEditorService;
    }

    public KafkaThread(KafkaService kafkaService, KafkaTopicService kafkaTopicService) {
        this.kafkaService = kafkaService;
        this.kafkaTopicService = kafkaTopicService;
    }
    @Override
    public void run() {
        while (true){
//            try {
//                Thread.sleep(10000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
            HashMap<String,Object> param = new HashMap<>();
            param.put("status","true");
            param.put("type","get");
            List<Kafka> byNamedParamList = kafkaService.findByNamedParamList(param);
            Kafka kafka = byNamedParamList.get(0);
            if(props!=null&&props.getProperty("bootstrap.servers").equals(kafka.getKafkaUrl())){
                return;
            }
            if(props!=null) props.clear();
            props = new Properties();
            props.put("bootstrap.servers", kafka.getKafkaUrl());
            props.put("group.id", "my-group1");
            props.put("auto.offset.reset", "earliest");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            kafkaTopicList = kafkaTopicService.findTopicByParamList(param);
            if(consumer!=null)consumer.close();
            consumer = new KafkaConsumer<>(props);
            consumer.subscribe(kafkaTopicList);
        }
    }
}
